#include	"../include/Redis.h"
#include	"../include/Logger.h"
#include	"../include/IOListener.h"

#include	"hiredis/hiredis.h"
#include	"hiredis/async.h"

#include	<cstdarg>
#include	<cstring>
#include	<thread>

#if defined(_WIN32)
#	include		<WinSock2.h>
#endif

struct redisAsyncListener : public IOListener {
	redisAsyncContext * ac;
	bool reading, writing;

	virtual void OnReadable() override { if (reading) redisAsyncHandleRead(ac); }
	virtual void OnWritable() override { if (writing) redisAsyncHandleWrite(ac); }
};

static void __redisAsyncAddRead(void * data) {
	redisAsyncListener * listener = (redisAsyncListener *)data;
	if (!listener->reading) listener->reading = true;
}

static void __redisAsyncDelRead(void * data) {
	redisAsyncListener * listener = (redisAsyncListener *)data;
	if (listener->reading) listener->reading = false;
}

static void __redisAsyncAddWrite(void * data) {
	redisAsyncListener * listener = (redisAsyncListener *)data;
	if (!listener->writing) listener->writing = true;
}

static void __redisAsyncDelWrite(void * data) {
	redisAsyncListener * listener = (redisAsyncListener *)data;
	if (listener->writing) listener->writing = false;
}

static void __redisAsyncCleanUp(void * data) {
	redisAsyncListener * listener = (redisAsyncListener *)data;
	listener->UnbindIO();
	delete listener;
}

static int redisAsyncAttach(redisAsyncContext *ac) {
	if (ac->ev.data != NULL) return REDIS_ERR;

	redisAsyncListener * listener = new redisAsyncListener();
	listener->ac = ac;
	listener->reading = false;
	listener->writing = true;

	ac->ev.addRead = __redisAsyncAddRead;
	ac->ev.delRead = __redisAsyncDelRead;
	ac->ev.addWrite = __redisAsyncAddWrite;
	ac->ev.delWrite = __redisAsyncDelWrite;
	ac->ev.cleanup = __redisAsyncCleanUp;
	ac->ev.data = listener;

	listener->BindIO(ac->c.fd, IO_READWRITE);
	return REDIS_OK;
}

namespace Redis {

	//////////////////////////////////////////////////////////////////////////
	/// Client status
	//////////////////////////////////////////////////////////////////////////
	enum Status {
		RS_Disconnected = 0,
		RS_Connecting,
		RS_Running,
		RS_Stoping,
		RS_Stopped
	};

	//////////////////////////////////////////////////////////////////////////
	/// IMPLEMENT OF Redis::Reply
	//////////////////////////////////////////////////////////////////////////
	Reply::~Reply() {
		if (_needfree && _reply) freeReplyObject(_reply);
	}

	Reply::operator bool() const {
		if (!_reply) return false;
		if (_reply->type == REDIS_REPLY_ERROR) return false;
		if (_reply->type == REDIS_REPLY_STATUS) return strncmp(_reply->str, "OK", 2) == 0;
		return true;
	}

	int Reply::Type() const {
		return _reply->type;
	}

	bool Reply::IsString() const {
		return _reply->type == REDIS_REPLY_STRING;
	}

	bool Reply::IsArray() const {
		return _reply->type == REDIS_REPLY_ARRAY;
	}

	bool Reply::IsInt() const {
		return _reply->type == REDIS_REPLY_INTEGER;
	}

	bool Reply::IsNil() const {
		return _reply->type == REDIS_REPLY_NIL;
	}

	const char * Reply::AsError() const {
		if (!_reply) return "NO REPLY";
		return _reply->str;
	}

	std::string Reply::AsString() const {
		return _reply->str ? std::string(_reply->str, _reply->len) : "";
	}

	int64_t Reply::AsInt() const {
		return (int64_t)(_reply->integer);
	}

	size_t Reply::Size() const {
		if (_reply->type != REDIS_REPLY_ARRAY) return 0;
		return _reply->elements;
	}

	Reply Reply::Element(size_t n) const {
		if (_reply->type != REDIS_REPLY_ARRAY || n > _reply->elements) return Reply(nullptr, false);
		return Reply(_reply->element[n], false);
	}

	//////////////////////////////////////////////////////////////////////////
	/// IMPLEMENT OF Redis::Simple
	//////////////////////////////////////////////////////////////////////////

	bool Simple::Connect(const std::string & host, int port) {
		if (_ctx) redisFree(_ctx);

		struct timeval wait;
		wait.tv_sec = 3;
		wait.tv_usec = 0;

		_ctx = redisConnectWithTimeout(host.c_str(), port, wait);
		if (!_ctx) {
			LOG_ERR("Failed to connect redis : alloc");
			return false;
		} else if (_ctx->err) {
			LOG_ERR("Failed to connect redis : %s", _ctx->errstr);
			redisFree(_ctx);
			_ctx = nullptr;
			return false;
		}

		LOG_INFO("Connected redis ... [OK] ");
		return true;
	}

	void Simple::Disconnect() {
		if (_ctx) redisFree(_ctx);
		_ctx = nullptr;
	}

	Reply Simple::Command(const char * cmd, ...) {
		if (!_ctx) return Reply(nullptr, false);

		va_list args;
		va_start(args, cmd);
		redisReply * reply = (redisReply *)redisvCommand(_ctx, cmd, args);
		va_end(args);

		if (!reply) {
			LOG_WARN("Send command to redis failed : %s", _ctx->errstr);

			if (_ctx->err == REDIS_ERR_EOF || _ctx->err == REDIS_ERR_IO) {
				if (redisReconnect(_ctx) != REDIS_OK) {
					LOG_WARN("Reconnected redis ... [Failed]");
				} else {
					LOG_INFO("Reconnected redis ... [OK]");
				}
			}
		}
		
		return Reply(reply, true);	
	}

	//////////////////////////////////////////////////////////////////////////
	/// IMPLEMENT OF Redis::Client
	//////////////////////////////////////////////////////////////////////////

	Client::Client()
		: _ctx(nullptr)
		, _status(RS_Stopped)
		, _cbid(1)
		, _callbacks() {
	}

	Client::~Client() {
		Disconnect();
	}

	bool Client::Connect(const std::string & host, int port) {
		if (_status != RS_Disconnected && _status != RS_Stopped) return false;
		if (_ctx) redisAsyncFree(_ctx);

		_status = RS_Connecting;
		_ctx = redisAsyncConnect(host.c_str(), port);
		if (!_ctx) {
			LOG_ERR("Failed to connect redis : alloc");
			_status = RS_Stopped;
			return false;
		} else if (_ctx->err) {
			LOG_ERR("Failed to connect redis : %s", _ctx->errstr);
			redisAsyncFree(_ctx);
			_status = RS_Stopped;
			_ctx = nullptr;
			return false;
		}

		_ctx->data = this;
		_host = host;
		_port = port;

		redisAsyncAttach(_ctx);

		redisAsyncSetConnectCallback(_ctx, [](const redisAsyncContext * ctx, int status) {
			Client * c = (Client *)ctx->data;
			if (status != REDIS_OK) {
				c->_status = RS_Stopped;
				LOG_ERR("Failed to connect redis : %d", status);
			} else {
				c->_status = RS_Running;
				LOG_INFO("Connected redis ... [OK]");
			}
		});

		redisAsyncSetDisconnectCallback(_ctx, [](const redisAsyncContext * ctx, int status) {
			Client * c = (Client *)ctx->data;
			if (c->_status != RS_Stoping) {
				c->_status = RS_Disconnected;
				LOG_WARN("Disconnected to redis server ...");

				std::thread t([&]() { c->Connect(c->_host, c->_port); });
				t.detach();
			} else {
				c->_status = RS_Stopped;
			}
		});

		redisAsyncListener * listener = (redisAsyncListener *)(_ctx->ev.data);
		listener->WaitIO(IO_WRITE);
		return _status == RS_Running;
	}

	void Client::Disconnect() {
		if (_status == RS_Stopped) return;

		_status = RS_Stoping;
		redisAsyncDisconnect(_ctx);

		_ctx = nullptr;
		_status = RS_Stopped;
		_callbacks.clear();
	}

	bool Client::Command(const ReplyHandler & callback, const char * cmd, ...) {
		if (_status != RS_Running) return false;

		va_list args;
		va_start(args, cmd);

		uint64_t seq = _cbid++;
		_callbacks.insert(std::pair<uint64_t, ReplyHandler>(seq, callback));

		int state = redisvAsyncCommand(_ctx, [](redisAsyncContext * ac, void * data, void * userdata) {
			Client * c = (Client *)ac->data;
			redisReply * reply = (redisReply *)data;
			uint64_t seqid = (uint64_t)(userdata);

			Reply r(reply, false);
			auto it = c->_callbacks.find(seqid);
			if (it != c->_callbacks.end()) it->second(r);
			c->_callbacks.erase(it);
		}, (void *)(uintptr_t)(seq), cmd, args);

		va_end(args);

		if (state != REDIS_OK) {
			LOG_ERR("Redis::Client::Command[Async way] failed : %d", state);
			return false;
		}

		return true;
	}
}